//package com.light.dp.flink.canal;
//
//import com.alibaba.otter.canal.client.CanalConnector;
//import com.alibaba.otter.canal.client.CanalConnectors;
//import com.alibaba.otter.canal.protocol.CanalEntry;
//import com.alibaba.otter.canal.protocol.Message;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.PostConstruct;
//import java.net.InetSocketAddress;
//import java.time.LocalDateTime;
//import java.util.List;
//
//@Component
//public class CanalMySqlSourceExample {
//    private int counter = 0;
//    @PostConstruct
//    public void init() throws Exception {
//        new Thread(() -> {
//            InetSocketAddress inetSocketAddress = new InetSocketAddress("192.168.1.72", 11111);
//            CanalConnector connector = CanalConnectors.newSingleConnector(inetSocketAddress, "city-supervise",
//                    "canal", "canal");
//            try {
//                // 连接到 Canal 服务器
//                connector.connect();
//                // 订阅感兴趣的数据库表，这里订阅所有表
//                connector.subscribe("city-supervise-service.quick_chat_msg");
//                // 回滚到未进行 ack 的地方，下次 fetch 时可以从最后一个没有 ack 的地方开始拿
//                connector.rollback();
//
//                while (true) {
//                    // 从 Canal 服务器获取指定数量的变更数据
//                    Message message = connector.get(100);
//                    long batchId = message.getId();
//                    int size = message.getEntries().size();
//                    //empty message
//                    if (batchId == -1 || size == 0) {
//                    } else {
////                        System.out.println("--------canal message ：" + message);
//                        // 计数器递增
//                        counter++;
//                        // 判断计数器是否达到 100000
//                        if (counter == 100000) {
//                            // 打印时间戳
//                            System.out.println("canal Reached 100000 messages at: " + LocalDateTime.now());
//                        }
//                    }
//
//                }
//            } finally {
//                // 关闭连接
//                connector.disconnect();
//            }
//        }).start();
//
//    }
//
//    private static void printColumn(List<CanalEntry.Column> columns) {
//        for (CanalEntry.Column column : columns) {
//            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
//        }
//    }
//}
